/*
 * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.otter.manager.biz.common.arbitrate;

import com.alibaba.otter.manager.biz.config.channel.ChannelService;
import com.alibaba.otter.manager.biz.config.node.NodeService;
import com.alibaba.otter.shared.arbitrate.impl.config.ArbitrateConfig;
import com.alibaba.otter.shared.arbitrate.impl.config.ArbitrateConfigRegistry;
import com.alibaba.otter.shared.common.model.config.ConfigException;
import com.alibaba.otter.shared.common.model.config.channel.Channel;
import com.alibaba.otter.shared.common.model.config.node.Node;
import com.alibaba.otter.shared.common.model.config.pipeline.Pipeline;
import com.alibaba.otter.shared.common.utils.cache.RefreshMemoryMirror;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.OtterMigrateMap;
import lombok.Setter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * manager下的基于db查询的{@linkplain ArbitrateConfig}实现
 *
 * @author jianghang 2011-11-3 上午11:09:24
 * @version 4.0.0
 */
@DependsOn("zookeeperClient")
@Component
public class ArbitrateConfigImpl implements ArbitrateConfig, InitializingBean, SmartInitializingSingleton {

    private static final Long DEFAULT_PERIOD = 60 * 1000L;
    private Long timeout = DEFAULT_PERIOD;
    private RefreshMemoryMirror<Long, Channel> channelCache;
    private LoadingCache<Long, Long> channelMapping;

    @Setter
    private ChannelService channelService;

    @Setter
    private NodeService nodeService;

    private RefreshMemoryMirror<Long, Node> nodeCache;

    public ArbitrateConfigImpl() {
        // 注册自己到arbitrate模块
        ArbitrateConfigRegistry.registry(this);
    }

    public Node currentNode() {
        return null;
    }

    public Node findNode(Long nid) {
        return nodeCache.get(nid);
    }

    public Channel findChannel(Long channelId) {
        return channelCache.get(channelId);
    }

    public Channel findChannelByPipelineId(Long pipelineId) {
        Long channelId = channelMapping.getUnchecked(pipelineId);
        return channelCache.get(channelId);
    }

    public Pipeline findOppositePipeline(Long pipelineId) {
        Long channelId = channelMapping.getUnchecked(pipelineId);
        Channel channel = channelCache.get(channelId);
        List<Pipeline> pipelines = channel.getPipelines();
        for (Pipeline pipeline : pipelines) {
            if (!pipeline.getId().equals(pipelineId)) {// 这里假定pipeline只有两个
                return pipeline;
            }
        }

        return null;
    }

    public Pipeline findPipeline(Long pipelineId) {

        Long channelId = channelMapping.getUnchecked(pipelineId);
        Channel channel = channelCache.get(channelId);
        List<Pipeline> pipelines = channel.getPipelines();
        for (Pipeline pipeline : pipelines) {
            if (pipeline.getId().equals(pipelineId)) {
                return pipeline;
            }
        }

        throw new ConfigException("no pipeline for pipelineId[" + pipelineId + "]");
    }

    public void afterPropertiesSet() {
        // 获取一下nid变量
        channelMapping = OtterMigrateMap.makeComputingMap(pipelineId -> {
            // 处理下pipline -> channel映射关系不存在的情况
            Channel channel = channelService.findByPipelineId(pipelineId);
            if (channel == null) {
                throw new ConfigException("No Such Channel by pipelineId[" + pipelineId + "]");
            }

            updateMapping(channel, pipelineId);// 排除下自己
            channelCache.put(channel.getId(), channel);// 更新下channelCache
            return channel.getId();

        });

        channelCache = new RefreshMemoryMirror<>(timeout, (key, oldValue) -> {
            Channel channel = channelService.findById(key);
            if (channel == null) {
                // 其他情况直接返回内存中的旧值
                return oldValue;
            } else {
                updateMapping(channel, null);// 排除下自己
                return channel;
            }
        });

        nodeCache = new RefreshMemoryMirror<>(timeout, (key, oldValue) -> {
            Node node = nodeService.findById(key);
            if (node == null) {
                return oldValue;
            } else {
                return node;
            }
        });
    }

    private void updateMapping(Channel channel, Long excludeId) {
        Long channelId = channel.getId();
        List<Pipeline> pipelines = channel.getPipelines();
        for (Pipeline pipeline : pipelines) {
            if (!pipeline.getId().equals(excludeId)) {
                channelMapping.put(pipeline.getId(), channelId);
            }
        }
    }


    @Override
    public void afterSingletonsInstantiated() {
        this.channelService = ArbitrateConfigRegistry.getBean(ChannelService.class);
        this.nodeService = ArbitrateConfigRegistry.getBean(NodeService.class);
    }
}
